package org.budo.dubbo.protocol.async.repository.activemq;

import java.util.ArrayList;
import java.util.List;

import javax.jms.BytesMessage;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;

import org.budo.dubbo.protocol.async.repository.BudoAsyncInvocation;
import org.budo.dubbo.protocol.async.util.AsyncLogUtil;
import org.budo.dubbo.protocol.async.util.AsyncSerializeUtil;
import org.budo.dubbo.protocol.async.util.DestinationUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
import com.alibaba.dubbo.rpc.RpcInvocation;

/**
 * @author lmw
 */
public class ActiveMQAsyncRepository extends AbstractActiveMQAsyncRepository {
    private static final Logger log = LoggerFactory.getLogger(ActiveMQAsyncRepository.class);

    private static final int RECEIVE_TIMEOUT = 100; // 毫秒

    /**
     * 引用端调用方法时发消息
     */
    @Override
    public void sendMessageWhenInvoke(Invocation invocation) {
        String destinationName = DestinationUtil.getDestinationName(invocation.getInvoker());

        BudoAsyncInvocation budoAsyncInvocation = BudoAsyncInvocation.fromDubboInvocation(invocation);

        this.sendMessage_1(destinationName, budoAsyncInvocation, invocation.getInvoker());
    }

    @Override
    public void sendMessage(String destinationName, Object message) {
        this.sendMessage_1(destinationName, message, null);
    }

    private void sendMessage_1(String destinationName, Object message, Invoker<?> invoker) {
        try {
            BudoActiveMQSession activeMQSession = this.getSession(destinationName);

            byte[] requestBody = AsyncSerializeUtil.serializeAsyncInvocation(message, invoker);

            BytesMessage bytesMessage = activeMQSession.getSession().createBytesMessage();
            bytesMessage.writeBytes(requestBody);

            MessageProducer messageProducer = activeMQSession.producer();
            messageProducer.send(bytesMessage);

            if (log.isDebugEnabled()) {
                log.debug("#39 sendMessageWhenInvoke, bytesMessage=" + bytesMessage + ", destinationName=" + destinationName + ", invoker=" + invoker);
            }
        } catch (Throwable e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 提供端收到消息后调用方法
     * 
     * @see org.apache.activemq.ActiveMQMessageConsumer#receive(long)
     */
    @Override
    public Integer popMessageAndInvoke(Invoker<?> invoker) {
        String destinationName = DestinationUtil.getDestinationName(invoker);

        List<BytesMessage> messageReceived = new ArrayList<BytesMessage>();

        BudoActiveMQSession activeMQSession = null;
        try {
            activeMQSession = this.getSession(destinationName);
            MessageConsumer messageConsumer = activeMQSession.consumer();

            int messageCount = 0;
            while (true) { // 一条一条的 把消息拉完
                BytesMessage bytesMessage = (BytesMessage) messageConsumer.receive(RECEIVE_TIMEOUT);
                if (null == bytesMessage) {
                    if (log.isDebugEnabled()) {
                        log.debug("#80 bytesMessage is null, destinationName=" + destinationName + ", messageConsumer=" + messageConsumer);
                    }

                    break; // 返回空就跳出
                }

                // 读取消息内容
                byte[] messageBody = new byte[(int) bytesMessage.getBodyLength()];
                bytesMessage.readBytes(messageBody);

                BudoAsyncInvocation budoAsyncInvocation = AsyncSerializeUtil.deserializeAsyncInvocation(messageBody, invoker);

                if (log.isDebugEnabled()) {
                    log.debug("#93 message received, bytesMessage=" + bytesMessage + ", budoAsyncInvocation=" + budoAsyncInvocation);
                }

                // 如果消息延迟就打日志
                AsyncLogUtil.logMessageDelay(budoAsyncInvocation, destinationName, invoker, this);

                RpcInvocation invocation = BudoAsyncInvocation.toDubboInvocation(budoAsyncInvocation);
                invoker.invoke(invocation); // 执行方法

                // 暂不提交，finally 里提交
                messageReceived.add(bytesMessage);
                messageCount++;
            }

            return messageCount;
        } catch (Throwable e) {
            return this.onError(invoker, activeMQSession, e);
        } finally {
            this.acknowledgeMessages(messageReceived); // 提交消息
        }
    }

    /**
     * 方法调用成功后确认收到消息，默认 ActiveMQSession.INDIVIDUAL_ACKNOWLEDG 单条确认模式
     * https://blog.csdn.net/zhu_tianwei/article/details/46303535
     */
    private void acknowledgeMessages(List<BytesMessage> messages) {
        for (BytesMessage message : messages) {
            try {
                message.acknowledge();
            } catch (JMSException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 出错时的处理，可能需要重连
     */
    private Integer onError(Invoker<?> invoker, BudoActiveMQSession activeMQSession, Throwable e) {
        if ((e + "").contains("The Consumer is closed")) {
            log.error("#107 popMessageAndInvoke error, invoker=" + invoker + ", e=" + e);

            if (null != activeMQSession) {
                activeMQSession.setMessageConsumer(null); // 设为空，下次重新连接
            }

            return 0;
        }

        // 其他错误
        throw new RuntimeException(e);
    }

    @Override
    public void sendRawMessage(String destinationName, Object rawMessage) {
        throw new RuntimeException("#157 destinationName=" + destinationName);
    }
}